package com.atguigu.rabbitmq.day04;

import com.atguigu.rabbitmq.utils.RabbitMqUtils;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.ConfirmCallback;

import java.util.UUID;
import java.util.concurrent.ConcurrentLinkedDeque;
import java.util.concurrent.ConcurrentNavigableMap;
import java.util.concurrent.ConcurrentSkipListMap;

/*验证发布确认模式：1单个确认 2.批量确认 3.异步批量确认
* 通过比较每种模式所用的时间来判断哪种是最快的*/
public class ConfirmMessage {
    //批量发消息的个数
    public static final int message_count=1000;
    public static void main(String[] args) throws Exception {
        //1.单个确认
        //publishMessageIndividually();//发布1000条单独确认消息，用时为：552ms
        //2.批量确认
        //publishMessageBatch();//发布1000条批量发布确认消息，用时为：75ms
        //3.异步批量确认
        publishMessageAsync();//发布1000条异步发布确认消息，用时为：43ms
    }
    //1.单个确认
    public static void publishMessageIndividually()throws Exception{
        Channel channel = RabbitMqUtils.getChannel();
        //队列的声明
        String queueName = UUID.randomUUID().toString();
        channel.queueDeclare(queueName,false,false,false,null);
        //开启发布确认
        channel.confirmSelect();

        //开始时间
        long begin = System.currentTimeMillis();

        //批量发消息  单个发布确认
        for (int i = 0; i < message_count; i++) {
            String message=i+"";
            channel.basicPublish("",queueName,null,message.getBytes());
            //单个消息就马上进行发布确认
            boolean flag = channel.waitForConfirms();
            if (flag){
                System.out.println("消息发送成功！");
            }
        }

        //结束时间
        long end = System.currentTimeMillis();

        System.out.println("发布"+message_count+"条单独确认消息，用时为："+(end-begin)+"ms");
    }

    //批量发布确认
    public static void publishMessageBatch()throws Exception{
        Channel channel = RabbitMqUtils.getChannel();
        //队列的声明
        String queueName = UUID.randomUUID().toString();
        channel.queueDeclare(queueName,false,false,false,null);
        //开启发布确认
        channel.confirmSelect();

        //开始时间
        long begin = System.currentTimeMillis();

        //批量确认消息大小
        int batchSize=100;

        //批量发布消息 批量发布确认
        for (int i = 0; i < message_count; i++) {
            String message=i+"";
            channel.basicPublish("",queueName,null,message.getBytes());
            //判断达到100条数据的时候批量确认一次
            if (i%batchSize==0){
                //发布确认
                channel.waitForConfirms();
            }
        }



        //结束时间
        long end = System.currentTimeMillis();

        System.out.println("发布"+message_count+"条批量发布确认消息，用时为："+(end-begin)+"ms");
    }

    //异步发布确认
    public static void publishMessageAsync()throws Exception{
        Channel channel = RabbitMqUtils.getChannel();
        //队列的声明
        String queueName = UUID.randomUUID().toString();
        channel.queueDeclare(queueName,false,false,false,null);
        //开启发布确认
        channel.confirmSelect();

        /**
         * 线程安全有序的一个哈希表 适用于高并发的情况下
         *功能
         * 1.轻松地将序号与消息进行关联
         * 2.轻松地批量删除条目 只要给到序号
         * 3.支持高并发(多线程)
         */
        ConcurrentSkipListMap<Long,String> outStandingConfirms=new ConcurrentSkipListMap<>();//泛型第一个Long代表消息的标记

        //消息确认成功 回调函数
        /**
         * 参数
         * 1.参数的标记
         * 2.是否为批量确认
         */
        ConfirmCallback ackCallback=( deliveryTag,  multiple)->{
            if (multiple){//如果是批量的，就批量删除
                //2.删除掉已经确认的消息 剩下的就是未确认的消息
                ConcurrentNavigableMap<Long, String> confirmed = outStandingConfirms.headMap(deliveryTag);
                confirmed.clear();
            }else {
                outStandingConfirms.remove(deliveryTag);//如果不是批量的就直接删除当前的消息即可
            }


            System.out.println("确认的消息编号(标记)："+deliveryTag);
        };
        //消息确认失败 回调函数
        ConfirmCallback nackCallback=( deliveryTag,  multiple)->{
            //3.打印一下未确认的消息都有哪些
            String message = outStandingConfirms.get(deliveryTag);
            System.out.println("未确认的消息是："+message+",未确认的消息编号(标记)："+deliveryTag);
        };
        //准备消息的监听器 监听哪些消息成功了，哪些消息失败了
        /**
         * 参数
         * 1.监听哪些消息成功了
         * 2.监听哪些消息失败了
         */
        channel.addConfirmListener(ackCallback,nackCallback);//不想监听失败的把后者改为null即可


        //开始时间
        long begin = System.currentTimeMillis();
        //批量发送消息 不要手动确认消息
        for (int i = 0; i < message_count; i++) {
            String message=i+"";
            channel.basicPublish("",queueName,null,message.getBytes());
            //1.此处记录下所有要发送的消息 消息的总和
            outStandingConfirms.put(channel.getNextPublishSeqNo(),message);



        }

        //结束时间
        long end = System.currentTimeMillis();

        System.out.println("发布"+message_count+"条异步发布确认消息，用时为："+(end-begin)+"ms");
    }
}

/*单个确认发布：这是一种简单的确认方式，它是一种同步确认发布的方式，也就是发布一个消息之后只有它被确认发布，后续的消息才能继续发布,
waitForConfirmsOrDie(long)这个方法只有在消息被确认的时候才返回，如果在指定时间范围内这个消息没有被确认那么它将抛出异常。
这种确认方式有一个最大的缺点就是:发布速度特别的慢，因为如果没有确认发布的消息就会阻塞所有后续消息的发布，这种方式最多提供每秒不超过数百条发布消息的
吞吐量。当然对于某些应用程序来说这可能已经足够了。*/

/*批量发布确认：上面那种方式非常慢，与单个等待确认消息相比，先发布一批消息然后一起确认可以极大地提高吞吐量，当然这种方式的缺点就是:当发生故障导致发布出现问题
时，不知道是哪个消息出现问题了，我们必须将整个批处理保存在内存中，以记录重要的信息而后重新发布消息。当然这种方案仍然是同步的，也一样阻塞消息的发布*/

/*异步发布确认：不管是单个确认发布还是批量确认发布，都是必须确认以后才能进行其他消息的发布，而异步发布确认是你只管发就行了，想发多少发多少，哪条消息
* 出错了一会就会通知你，而不是必须确认后才能发布下一条消息，这就大大提高了效率*/
